package cn.itbeien.payment.message;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationRunner;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

/**
 * @author beien
 * @date 2024-03-22 10:28
 * Copyright© 2024 beien
 */
@Component
@Slf4j
public class DelayMessage {
    @Autowired
    private StreamBridge streamBridge;
   // @Bean
    public ApplicationRunner producerDelay() {
        return args -> {
            for (int i = 0; i < 100; i++) {
                String key = "KEY" + i;
                Map<String, Object> headers = new HashMap<>();
                headers.put(MessageConst.PROPERTY_KEYS, key);
                headers.put(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, i);
                // 设置延时等级1~10
                headers.put(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 4);
                Message<SimpleMsg> msg = new GenericMessage(new SimpleMsg("Delay RocketMQ " + i), headers);
                streamBridge.send("producer-out-0", msg);
            }
        };
    }

   // @Bean
    public Consumer<Message<SimpleMsg>> consumer() {
        return msg -> {
            log.info(Thread.currentThread().getName() + " Consumer Receive New Messages: " + msg.getPayload().getMsg());
        };
    }
}
